package cn.media

import java.sql.{Connection, DriverManager, PreparedStatement}
import java.util.Properties

import com.typesafe.config.{Config, ConfigFactory}
import org.apache.spark.broadcast.Broadcast
import org.apache.spark.rdd.RDD
import org.apache.spark.{SparkConf, SparkContext}
import org.apache.spark.sql.{DataFrame, SQLContext, SaveMode}

object mediaV1 {

    def main(args: Array[String]): Unit = {

      //校验参数个数
      if (args.length != 2) {
        println(
          """
            |cn.ad.tools.AppTypeRpt
            |参数：
            | appNameInputPath
            | logInputPath
          """.stripMargin)
        sys.exit()
      }
      //接受程序参数
      val Array(logInputPath, appNameInputPath) = args

      val conf: SparkConf = new SparkConf()
        .setMaster("local[*]")
        .setAppName(s"${this.getClass.getSimpleName}")

      val sc: SparkContext = new SparkContext(conf)

      val sqlsc: SQLContext = new SQLContext(sc)

      //读取AppName数据
      val appName: RDD[String] = sqlsc.sparkContext.textFile(appNameInputPath)

      //整理出appId和appName
      val appIDAndAppName: RDD[(String, String)] = appName.map(line => {
        val fileds: Array[String] = line.split("\t",-1)
        val appID = fileds(4)
        val appName = fileds(1)
        (appID, appName)
      })

      //收集到driver端
      val idAndName: Array[(String, String)] = appIDAndAppName.collect()

      //广播(必须使用sparkcontext)
      //将广播变量的引用返回到Driver端
      val broadcast: Broadcast[Array[(String, String)]] = sqlsc.sparkContext.broadcast(idAndName)

      //读取数据
      val parquet: DataFrame = sqlsc.read.parquet(logInputPath)
      //注册表
      parquet.registerTempTable("t_log")
      //自定义方法
      sqlsc.udf.register("appName", (appid: String, appName: String) => {
        var appname = appName
        if (appname != null && !"".equals(appname)){
          appname
        }else{
          //使用广播变量的引用
          val appIdAndName: Array[(String, String)] = broadcast.value
          for(tuple <- appIdAndName){
            if(tuple._1.equals(appid)){
              appname = tuple._2
            }
          }
        }
        appname
      })
      //
      val result: DataFrame = sqlsc.sql(
        //appid: String,13
        //appname: String,14
        """
          |select
          |appName(appid, appname) appName
          |,sum(case when requestmode = 1 and processnode >= 1 then 1 else 0 end)as origintNum
          |,sum(case when requestmode = 1 and processnode >= 2 then 1 else 0 end)as validNum
          |,sum(case when requestmode = 1 and processnode = 3 then 1 else 0 end)as adRequestNum
          |,sum(case when iseffective = 1 and isbilling = 1 and isbid = 1 and adorderid !=0 then 1 else 0 end)as bidNum
          |,sum(case when iseffective = 1 and isbilling = 1 and iswin = 1 then 1 else 0 end)as bidSuccessNum
          |,sum(case when requestmode = 2 and iseffective = 1 then 1 else 0 end)as showNum
          |,sum(case when requestmode = 3 and iseffective = 1 then 1 else 0 end)as pointNum
          |,sum(case when iseffective = 1 and isbilling = 1 and iswin = 1 then winprice*1.0/1000 else 0.0 end)as costMoney
          |,sum(case when iseffective = 1 and isbilling = 1 and iswin = 1 then adpayment*1.0/1000 else 0.0 end)as payMoney
          |from t_log
          |group by
          |appName(appid, appname)
        """.stripMargin)

      //JDBC配置
      val load: Config = ConfigFactory.load()
      val prop = new Properties()
      prop.put("user","root")
      prop.put("password","123456")
      result.write.mode(SaveMode.Ignore).jdbc(load.getString("jdbc.url"),load.getString("jdbc.tableName"),
        prop)

      sc.stop()
    }

}
